//消费者
package mq

import (
	"github.com/streadway/amqp"
	"log"
	"sync"
)

//ConsumerCallBack 实际的消费回调函数
type ConsumerCallBack func(msg []byte) bool

//标志是否已经开始消费，若开始则无法重复开始
var isStarted bool

//消费开始锁
var startLocker sync.Mutex

//用于控制停止消费
var exitChan = make(chan bool)

//StartConsumer:开始消费
//若返回DisConnectError 则需要自行处理重连
func StartConsumer(queueName, consumerName string, cb ConsumerCallBack) error {
	if innerChannel == nil {
		return DisConnectError
	}

	if isStarted {
		log.Println("StartConsumer已执行过")
		return nil
	}

	startLocker.Lock()
	if isStarted {
		log.Println("StartConsumer已执行过")
		return nil
	}
	var deliverChan <-chan amqp.Delivery
	if deliverChan = getDeliveryChan(queueName, consumerName); deliverChan == nil {
		startLocker.Unlock()
		return DisConnectError
	}
	isStarted = true
	startLocker.Unlock()
	defer innerChannel.Close()
	overChan := make(chan bool)      //用于防止StartConsumer退出
	exceptionChan := make(chan bool) //用于处理因为断连导致的消费退出，需要重连
	go func() {
		realConsume(cb, deliverChan, overChan, exceptionChan)
	}()

	log.Println("开始进行消费")
	select {
	case <-overChan:
		log.Println("消费结束")
		return nil
	case <-exceptionChan:
		log.Println("消费异常，需要重试")
		isStarted = false
		return DisConnectError
	}

}

func getDeliveryChan(queueName, consumerName string) <-chan amqp.Delivery {
	//autoAck为false，不自动应答，需要手动应答
	//exclusive:false,公平消费，非唯一消费者
	//noLocal：false rabbitmq不支持该设置
	//noWait:false 等待消息
	if innerChannel == nil {
		return nil
	}
	deliverChan, err := innerChannel.Consume(queueName, consumerName, false, false, false, false, nil)
	if err != nil {
		log.Panicf("设置开始消费异常：%v", err.Error())
		return nil
	}
	return deliverChan
}

func realConsume(cb ConsumerCallBack, deliverChan <-chan amqp.Delivery, overChan chan bool, exceptionChan chan bool) {
	for {
		select {
		case deliver, ok := <-deliverChan:
			if !ok {
				log.Println("接收异常，管道被关闭")
				exceptionChan <- true
				return
			}
			success := cb(deliver.Body)
			if success {
				//消费成功，确认成功，非批量确认 单条确认该消息
				log.Println("消费成功，进行确认")
				deliver.Ack(false)
			} else {
				if deliver.Body != nil {
					log.Println("消费失败，进入死信队列")
					//消费失败 ，使消息进入死信队列，由死信队列进行重试
					deliver.Nack(false, false)
				}
			}
		case <-exitChan:
			overChan <- true
			return
		}
	}
}

//停止消费
func StopConsumer() {
	exitChan <- true
}
